package main

import (
	"bufio"
	"bytes"
	"fmt"
	"github.com/golang/protobuf/proto"
	mesos "mesos/v1/master"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

func main() {
	mesosUrl := "http://10.211.55.6:5050/api/v1"

	if len(os.Args) > 1 && len(strings.TrimSpace(os.Args[1])) > 0 {
		mesosUrl = strings.TrimSpace(os.Args[1])
	}

	subscribedCall := &mesos.Call{Type: mesos.Call_SUBSCRIBE.Enum()}

	requestPayload, _ := proto.Marshal(subscribedCall)

	fmt.Printf("%s we subscribe the mesos: %s\n", time.Now().Format("2006-01-02 15:04:05"), mesosUrl)

	req, _ := http.NewRequest("POST", mesosUrl, bytes.NewReader(requestPayload))

	req.Header.Set("Content-Type", "application/x-protobuf")
	req.Header.Set("Accept", "application/x-protobuf")
	req.Header.Set("Message-Accept", "application/recordio")

	res, _ := http.DefaultClient.Do(req)

	defer func() {
		if res != nil {
			res.Body.Close()
		}
	}()

	buf := bufio.NewReaderSize(res.Body, 104857600) // 100MiB Just a Example

	recordQueue := make(chan []byte, 10)

	go recordProcess(recordQueue)

	for i := 0; i < 10; i++ {

		record, err := readRecord(buf)
		if err != nil {
			fmt.Println("read record failed: ", err)
			break
		}

		fmt.Printf("\n%s get the %d record, length: %d\n",
			time.Now().Format("2006-01-02 15:04:05"), i, len(record))
		recordQueue <- record

		time.Sleep(100 * time.Millisecond) // sleep 0.1s
	}

	close(recordQueue)

}

// goroutine process each record
func recordProcess(ch chan []byte) {
	for {
		record, ok := <-ch
		if !ok {
			break
		}
		handleEventRecord(record)
	}
}

func handleEventRecord(data []byte) {

	// parse record binary data to event object
	event := &mesos.Event{}
	err := proto.Unmarshal(data, event)
	if err != nil {
		fmt.Println("Parse protobuf data Failed: ", err)
		return
	}

	// do something for this event
	et := event.GetType().String()

	fmt.Println("This is a " + et + " Event")

	switch event.GetType() {
	case mesos.Event_SUBSCRIBED:
		showSubscribedEvent(event)
		break
	case mesos.Event_TASK_ADDED:
		showTaskAddedEvent(event)
		break
	case mesos.Event_TASK_UPDATED:
		showTaskUpdatedEvent(event)
		break
	}

}

func showSubscribedEvent(e *mesos.Event) {
	s := e.GetSubscribed().GetGetState()
	fmt.Printf("The cluster has %v agents, %v frameworks, %v tasks\n",
		len(s.GetGetAgents().GetAgents()),
		len(s.GetGetFrameworks().GetFrameworks()),
		len(s.GetGetTasks().GetTasks()))
}

func showTaskAddedEvent(e *mesos.Event) {
	task := e.GetTaskAdded().GetTask()
	fmt.Printf("A new task %v was %v \n", *(task.TaskId.Value),
		task.State)
}

func showTaskUpdatedEvent(e *mesos.Event) {
	update := e.GetTaskUpdated()
	fmt.Printf("The task %v was %v \n", *(update.GetStatus().TaskId.Value),
		update.GetState())
}

func readRecord(buf *bufio.Reader) ([]byte, error) {

	// first we read the first part of the record, it ends with a LF
	head, err := buf.ReadBytes('\n')
	if err != nil {
		return nil, err
	}

	length, err := strconv.Atoi(strings.TrimSpace(string(head)))
	if err != nil {
		return nil, err
	}

	// then we read the third part of the record, it has a mixed size
	data, err := buf.Peek(length)

	if err == nil {
		buf.Discard(length)
	}

	return data, err
}
